package org.jeecg.modules.mqtt;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.Resource;

/**
 * @program: simulator-b20240918
 * @description:
 * @author: renzhg
 * @create: 2024-10-28 11:13
 **/
@Configuration
public class MQTTProduceConfig {

    /**
     * mqtt配置类
     */
    @Resource
    private MQTTConfig mqttConfig;

    /**
     * 客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息
     */
    private static final byte[] WILL_DATA;

    static {
        WILL_DATA = "offline".getBytes();
    }

    /**
     * 出站直连通道
     *
     * @return MessageChannel
     */
    @Bean("mqttOut")
    public MessageChannel mqttOutBoundChannel() {
        return new DirectChannel();
    }

    /**
     * 创建MqttPahoClientFactory 设置MQTT的broker的连接属性
     *
     * @return MqttPahoClientFactory
     */
    @Bean
    public MqttPahoClientFactory outClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        String[] hosts = mqttConfig.getHost();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(hosts);
        options.setUserName(mqttConfig.getUsername());
        options.setPassword(mqttConfig.getPassword().toCharArray());
        options.setConnectionTimeout(mqttConfig.getTimeout());
        options.setKeepAliveInterval(mqttConfig.getKeepalive());
        options.setCleanSession(mqttConfig.getCleanSession());
        options.setAutomaticReconnect(mqttConfig.getAutomaticReconnect());
        options.setWill("willTopic", WILL_DATA, 2, false);
        factory.setConnectionOptions(options);
        return factory;
    }

    /**
     * 出站
     *
     * @return MessageHandler
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOut")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttConfig.getClientId() + "_producer", outClientFactory());
        //如果设置成true，即异步，发送消息时将不会阻塞。
        messageHandler.setAsync(true);
        //设置默认QoS
        messageHandler.setDefaultQos(mqttConfig.getQos());
        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();
        //发送默认按字节类型发送消息
        //defaultPahoMessageConverter.setPayloadAsBytes(true);
        messageHandler.setConverter(defaultPahoMessageConverter);
        return messageHandler;
    }
}
